package com.hetao.work;

import com.hetao.work.enums.MethodEnum;
import com.hetao.work.model.HttpResult;
import com.hetao.work.model.SingleTask;
import com.hetao.work.model.Task;
import com.hetao.work.model.SingleTaskResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class WorkProcess {

    protected Logger log = LoggerFactory.getLogger(getClass());
//    public void run(Task task) throws InterruptedException {
//        ExecutorService threadPool = Executors.newFixedThreadPool(task.getConnectCount());
//        for (int i=0; i<task.getConnectCount(); i++) {
//            SingleTask singleTask = new SingleTask();
//            singleTask.setNumber(i);
//            singleTask.setUrl(task.getUrl());
//            singleTask.setSendCount(task.getSendCount());
//            singleTask.setUrl(task.getUrl());
//
//            threadPool.execute(new SingleTaskRun(singleTask));
//
//        }
//
//        // 所有任务执行完成且等待队列中也无任务关闭线程池
//        threadPool.shutdown();
//        threadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
//        log.info("========================");
//        log.info("=======主线程执行完了======");
//        log.info("========================");
//    }
    public void run(Task task) throws InterruptedException, ExecutionException {
        task.setStart(Instant.now());

        List<SingleTask> list = IntStream.range(0, task.getConnectCount()).mapToObj((no)->{
            SingleTask singleTask = new SingleTask();
            singleTask.setNumber(no);
            singleTask.setUrl(task.getUrl());
            singleTask.setSendCount(task.getSendCount());
            singleTask.setUrl(task.getUrl());
            return singleTask;
        }).collect(Collectors.toList());

        ExecutorService threadPool = Executors.newFixedThreadPool(task.getConnectCount());
        List<SingleTaskResult> resultList = new ArrayList<>(task.getConnectCount()*task.getSendCount());
        List<CompletableFuture<Void>> futureList = list.parallelStream().map(singleTask -> {
            CompletableFuture<Void> future = CompletableFuture.runAsync(()->{

                HttpClient httpClient = new HttpClient();
                List<SingleTaskResult> sResults = new ArrayList<>();
                IntStream.range(0, singleTask.getSendCount()).forEach((i)->{
                    log.info("no: {} sendCount: {}", singleTask.getNumber(), i);
                    sResults.add(runSingleTask(httpClient, singleTask));
                });
                resultList.addAll(sResults);
            }, threadPool);
            return future;
        }).collect(Collectors.toList());

        futureList.forEach((future)-> {
            try {
                future.get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });

        log.info("========================");
        log.info("=======主线程执行完了======");
        log.info("========================");

        task.setEnd(Instant.now());

        log.info("总共花 {}s  连接数:{}个  每个请求: {}次", task.getEnd().getEpochSecond()- task.getStart().getEpochSecond(),
                task.getConnectCount(), task.getSendCount());


        statResult(task, resultList);
//        log.info("result: {}", JSON.toJSONString(resultList));
    }

    public void statResult(Task task, List<SingleTaskResult> resultList) {
//        long start = task.getStart().toEpochMilli()%1000;
//        long end = task.getEnd().toEpochMilli()%1000+1000;
//        long idx = start;
//        while(idx < end) {
//
//
//
//
//            idx += 1000;
//        }

        Map<Long, List<SingleTaskResult>> tmpMap = resultList.parallelStream().peek((item)-> item.setStatTime(item.getStart().getEpochSecond())).
                collect(Collectors.groupingBy(SingleTaskResult::getStatTime));

        TreeMap<Long, List<SingleTaskResult>> timeMap = new TreeMap<>(tmpMap);
        timeMap.forEach((k, v) -> {
//            LocalDateTime date = LocalDateTime.ofInstant(Instant.ofEpochSecond(k), ZoneId.systemDefault());
            LongSummaryStatistics statistics = v.stream().mapToLong(SingleTaskResult::getCost).summaryStatistics();

            log.info("date: {} max:{} min:{} average:{} count:{}", k-task.getStart().getEpochSecond(), statistics.getMax(), statistics.getMin(),
                    statistics.getAverage(), statistics.getCount());
        });

    }

    public SingleTaskResult runSingleTask(HttpClient httpClient, SingleTask task) {
        Instant start = Instant.now();
        HttpResult result;
        if (MethodEnum.POST.equals(task.getMethod())) {
            result = httpClient.post(task.getUrl(), null);
        } else {
            result = httpClient.get(task.getUrl());
        }

        Instant end = Instant.now();
        log.info("耗时：" + Duration.between(start, end).toMillis() + "ms");

        return SingleTaskResult.builder().cost(Duration.between(start, end).toMillis()).start(start).end(end).
                httpResult(result).build();
    }
}
